package bblnamed

import (
	"bblcloud/bblvolume"
	"encoding/json"
	"io/ioutil"
	"net"
	"net/http"
	"runtime/debug"
	"strconv"
	"strings"
	"time"
)

func NewClusterMux(cm ClusterManager, ms MetadataSystem) *ClusterMux {
	return &ClusterMux{ms: ms, cm: cm}
}

type ClusterMux struct {
	ms MetadataSystem
	cm ClusterManager
}

func notFound(rw http.ResponseWriter, req *http.Request) {
	http.NotFound(rw, req)
}
func badRequest(rw http.ResponseWriter, req *http.Request) {
	http.Error(rw, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
}
func internalServerError(rw http.ResponseWriter, req *http.Request) {
	http.Error(rw, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
}

func forbidden(rw http.ResponseWriter, req *http.Request) {
	http.Error(rw, http.StatusText(http.StatusForbidden), http.StatusForbidden)
}

type HandlerFunc func(mux *ClusterMux, rw http.ResponseWriter, req *http.Request)

var (
	op = map[string]HandlerFunc{
		"READ":          do_read,
		"WRITE":         do_write,
		"STAT":          do_stat,
		"RENAME":        do_rename,
		"LINK":          do_link,
		"UNLINK":        do_unlink,
		"DELETE":        do_delete,
		"TRUNCATE":      do_truncate,
		"ACK_WRITE":     do_ack_write,
		"ACK_APPEND":    do_ack_append,
		"ADD_VOLUME":    do_add_volume,
		"REMOVE_VOLUME": do_rm_volume,
		"ADD_CLIENT":    do_add_client,
		"REMOVE_CLIENT": do_rm_client,
		"JOIN":          do_join,
		"LEAVE":         do_leave,
		"GET_JOB":       do_get_job,
		"ACK_JOB":       do_ack_job,
		"CLUSTER_STAT":  do_cluster_stat,
	}

	op_valid = map[string]bool{
		"READ":          true,
		"WRITE":         true,
		"STAT":          true,
		"RENAME":        true,
		"LINK":          true,
		"UNLINK":        true,
		"DELETE":        true,
		"TRUNCATE":      true,
		"ACK_WRITE":     true,
		"ACK_APPEND":    true,
		"ADD_VOLUME":    false,
		"REMOVE_VOLUME": false,
		"ADD_CLIENT":    false,
		"REMOVE_CLIENT": false,
		"JOIN":          false,
		"LEAVE":         false,
		"GET_JOB":       false,
		"ACK_JOB":       false,
		"CLUSTER_STAT":  false,
	}
)

func (m *ClusterMux) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
	log.Debug(req.RemoteAddr, req.Header.Get("X-Forward-For"), req.Header.Get("Host"), req.Method, req.URL.String())
	defer func() {
		if e := recover(); e != nil {
			log.Error(string(debug.Stack()))
			internalServerError(rw, req)
		}
	}()
	rw.Header().Set("Server", server_name())
	if valid, exist := op_valid[req.Method]; exist && valid {
		addr, _, _ := net.SplitHostPort(req.RemoteAddr)
		if !m.cm.ValidGateway(addr) {
			forbidden(rw, req)
			return
		}
	}
	if handler, exist := op[req.Method]; exist {
		handler(m, rw, req)
	} else {
		badRequest(rw, req)
	}
}

type Stat struct {
	Size  int64 `json:"size"`
	CTime int64 `json:"ctime"`
	MTime int64 `json:"mtime"`
	UID   int   `json:"uid"`
	GID   int   `json:"gid"`
	Perm  int   `json:"perm"`
	Crc32 int64 `json:"crc32"`
}

func render_json(obj interface{}, rw http.ResponseWriter, req *http.Request) {
	b, err := json.Marshal(obj)
	if err != nil {
		log.Error(err)
		internalServerError(rw, req)
		return
	}
	rw.Header().Set("Content-Type", "application/json")
	rw.Write(b)
}
func do_stat(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	var meta Metadata
	var name = req.URL.Path
	meta, err := mux.ms.Find(name)
	if err != nil {
		log.Error(err)
		notFound(rw, req)
		return
	}
	stat := &Stat{Size: meta.Size(), CTime: meta.CreateTime(), MTime: meta.ModTime(), UID: meta.UID(), GID: meta.GID(), Perm: meta.Perm()}
	if meta.IsLink() {
		meta_o, _ := mux.ms.FindReal(meta.Link())
		if meta_o != nil {
			stat.Size = meta_o.Size()
		}
	}
	render_json(stat, rw, req)
}

func do_read(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	meta, err := mux.ms.FindReal(req.URL.Path)
	if err != nil {
		log.Error(err)
		notFound(rw, req)
		return
	}
	log.Debug("FindReal", meta)
	vols, err := mux.cm.FindVolume(meta.VolumeID())
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	if len(vols) == 0 {
		log.Error("vols not found")
		forbidden(rw, req)
		return
	}
	log.Debug("FindVolume", vols)
	ret := []string{}
	for _, vol := range vols {
		if vol.Status == ONLINE {
			ret = append(ret, vol.Address)
		}
	}
	if meta.Link() != req.URL.Path {
		rw.Header().Set("X-LINK-TO", meta.Link())
	}
	render_json(ret, rw, req)
}

func do_write(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	if _, err := mux.ms.Find(req.URL.Path); err == nil {
		//如果meta存在说明已经有对象存在，清理网关缓存
		go do_purge_gateway(mux, req)
	}
	vols, err := mux.cm.LocateVolume(req.URL.Path)
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	log.Debug("LocateVolume", vols)
	if vols == nil || len(vols) == 0 {
		log.Error("vol not found")
		forbidden(rw, req)
		return
	}

	meta, err := mux.ms.Create(req.URL.Path, 0)
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	log.Debug("create metadata", meta)
	ret := []string{}
	for _, vol := range vols {
		err = meta.AddVolumeID(vol.UUID)
		if err != nil {
			log.Error(err)
			forbidden(rw, req)
			return
		}
		if vol.Status == OFFLINE {
			log.Error("volume server ", vol.UUID, "is offline")
			forbidden(rw, req)
			return
		}
		ret = append(ret, vol.Address)
	}
	log.Debug("create metadata", meta)
	render_json(&ret, rw, req)

}

func do_purge_gateway(mux *ClusterMux, req *http.Request) {
	if gws, err := mux.cm.ListGateway(); err == nil {
		raddr, _, _ := net.SplitHostPort(req.RemoteAddr)
		for _, c := range gws {
			if !strings.HasPrefix(c.Address, raddr) {
				Purge(req.Header.Get("X-Host"), c.Address, req.URL.Path)
			}
		}
	}
}
func do_rename(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	meta, err := mux.ms.Find(req.URL.Path)
	if err != nil {
		log.Error(err)
		notFound(rw, req)
		return
	}
	go do_purge_gateway(mux, req)
	name := req.Header.Get("X-RENAME-TO")
	if name == "" {
		badRequest(rw, req)
		return
	}
	if name == req.URL.Path {
		badRequest(rw, req)
		return
	}
	to_meta, err := mux.ms.Find(name)
	if to_meta != nil {
		badRequest(rw, req)
		return
	}
	err = meta.SetName(name)
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	err = mux.ms.Remove(req.URL.Path)
	if err != nil {
		log.Error(err)
	}
}

func do_link(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	meta, err := mux.ms.Find(req.URL.Path)
	if err != nil {
		notFound(rw, req)
		return
	}
	name := req.Header.Get("X-LINK-TO")
	if name == "" {
		badRequest(rw, req)
		return
	}
	if name == req.URL.Path {
		badRequest(rw, req)
		return
	}
	to_meta, _ := mux.ms.Find(name)
	if to_meta != nil {
		badRequest(rw, req)
		return
	}
	to_meta, err = mux.ms.Create(name, 1)
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	err = to_meta.SetLink(meta.Name())
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
}

func do_unlink(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	meta, err := mux.ms.Find(req.URL.Path)
	if err != nil { //obj is exist
		log.Error(err)
		notFound(rw, req)
		return
	}
	if meta.IsObject() {
		log.Error(req.URL.Path, "not a link object")
		badRequest(rw, req)
		return
	}
	err = mux.ms.Remove(meta.Name())
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
}

func do_delete(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	meta, err := mux.ms.Find(req.URL.Path)
	if err != nil { //obj is exist
		log.Error(err)
		notFound(rw, req)
		return
	}
	go do_purge_gateway(mux, req)
	err = mux.ms.Remove(meta.Name())
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	if meta.IsObject() {
		vols, err := mux.cm.FindVolume(meta.VolumeID())
		if err != nil {
			log.Error(err)
			forbidden(rw, req)
			return
		}
		log.Debug("FindVolume", vols)
		for _, vol := range vols {
			if err = Delete(vol.Address, meta.Link()); err != nil {
				log.Error(err)
			}
		}
	}

}

func do_truncate(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	meta, err := mux.ms.FindReal(req.URL.Path)
	if err != nil { //obj is exist
		notFound(rw, req)
		return
	}
	go do_purge_gateway(mux, req)
	err = meta.SetSize(0)
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	if meta.IsObject() {
		vols, err := mux.cm.FindVolume(meta.VolumeID())
		if err != nil {
			log.Error(err)
			forbidden(rw, req)
			return
		}
		log.Debug("FindVolume", vols)
		for _, vol := range vols {
			if err = Trancate(vol.Address, meta.Link()); err != nil {
				log.Error(err)
			}
		}
	}
}

func do_ack_write(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	meta, err := mux.ms.Find(req.URL.Path)
	if err != nil { //obj is exist
		notFound(rw, req)
		return
	}
	x_written_size := req.Header.Get("X-WRITTEN-SIZE")
	if "" == x_written_size {
		badRequest(rw, req)
		return
	}
	size, err := strconv.ParseInt(x_written_size, 10, 64)
	if err != nil { //obj is exist
		log.Error(err)
		badRequest(rw, req)
		return
	}
	meta.SetSize(meta.Size() + size)
}

func do_ack_append(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	meta, err := mux.ms.Find(req.URL.Path)
	if err != nil { //obj is exist
		notFound(rw, req)
		return
	}
	x_written_size := req.Header.Get("X-WRITTEN-SIZE")
	if "" == x_written_size {
		badRequest(rw, req)
		return
	}
	size, err := strconv.ParseInt(x_written_size, 10, 64)
	if err != nil { //obj is exist
		log.Error(err)
		badRequest(rw, req)
		return
	}
	meta.SetSize(meta.Size() + size)
}

func do_add_volume(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	vol_id := req.URL.Path[1:]
	if vol_id == "" {
		badRequest(rw, req)
		return
	}
	if err := mux.cm.AddVolume(vol_id); err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
}

func do_rm_volume(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	vol_id := req.URL.Path[1:]
	if vol_id == "" {
		badRequest(rw, req)
		return
	}
	if err := mux.cm.RemoveVolume(vol_id); err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
}

func do_add_client(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	param := req.URL.Path[1:]
	if param == "" {
		badRequest(rw, req)
		return
	}
	gw := strings.Split(param, ",")
	if len(gw) < 2 {
		badRequest(rw, req)
		return
	}
	if err := mux.cm.AddGateway(gw[0], gw[1]); err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
}

func do_rm_client(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	addr := req.URL.Path[1:]
	if addr == "" {
		badRequest(rw, req)
		return
	}

	if err := mux.cm.RemoveGateway(addr); err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
}

func do_join(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	vol_id := req.URL.Path[1:]
	if vol_id == "" {
		badRequest(rw, req)
		return
	}

	b, err := ioutil.ReadAll(req.Body)

	if err != nil {
		log.Error(err)
		badRequest(rw, req)
		return
	}
	log.Debug(string(b))
	val := map[string]string{}
	err = json.Unmarshal(b, &val)
	if err != nil {
		log.Error(err)
		badRequest(rw, req)
		return
	}
	if v, exist := val["uuid"]; !exist || v == "" {
		badRequest(rw, req)
		return
	}
	if val["uuid"] != vol_id {
		badRequest(rw, req)
		return
	}
	vol := &Volume{UUID: vol_id}
	if v, exist := val["name"]; !exist || v == "" {
		badRequest(rw, req)
		return
	} else {
		vol.Name = v
	}
	if v, exist := val["tran_addr"]; !exist || v == "" {
		badRequest(rw, req)
		return
	} else {
		vol.Address = v
	}
	if v, exist := val["export"]; !exist || v == "" {
		badRequest(rw, req)
		return
	} else {
		vol.Export = v
	}

	if err = mux.cm.JoinVolume(vol); err != nil {
		log.Error(err)
		forbidden(rw, req)
	}
}

func do_leave(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	vol_id := req.URL.Path[1:]
	if vol_id == "" {
		badRequest(rw, req)
		return
	}
	vol, err := mux.cm.FindOneVolume(vol_id)
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	if vol == nil {
		forbidden(rw, req)
		return
	}
	mux.cm.LeaveVolume(vol_id)
}

func do_get_job(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	vol_id := req.URL.Path[1:]
	if vol_id == "" {
		badRequest(rw, req)
		return
	}
	vol, err := mux.cm.FindOneVolume(vol_id)
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	if vol == nil {
		forbidden(rw, req)
		return
	}
	jobs, err := mux.cm.LookupJob(vol_id, 5)
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	b, err := ioutil.ReadAll(req.Body)
	if err != nil {
		log.Error(err)
	} else {
		stat := bblvolume.VolStat{}
		json.Unmarshal(b, &stat)
		vol.Stat = &stat
		vol.HeartbeatTime = time.Now()
		vol.Status = ONLINE
	}

	render_json(jobs, rw, req)

}

func do_ack_job(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {

	vol_id := req.URL.Path[1:]
	if vol_id == "" {
		badRequest(rw, req)
		return
	}
	vol, err := mux.cm.FindOneVolume(vol_id)
	if err != nil {
		log.Error(err)
		forbidden(rw, req)
		return
	}
	if vol == nil {
		forbidden(rw, req)
		return
	}
	err = req.ParseForm()
	if err != nil {
		log.Error(err)
		badRequest(rw, req)
		return
	}
	id := req.FormValue("id")
	if err = mux.cm.RemoveJob(vol_id, id); err != nil {
		log.Error(err)
	}
}

func do_cluster_stat(mux *ClusterMux, rw http.ResponseWriter, req *http.Request) {
	stat := mux.cm.ClusterStat()
	var cstat struct {
		TotalSpace int64
		UserSpace  int64
		Status     int
		Version    int
		Volumes    []*Volume
		Clients    []*Gateway
	}
	cstat.TotalSpace = stat.TotalSpace()
	cstat.UserSpace = stat.UsedSpace()
	cstat.Status = stat.Version()
	cstat.Status = stat.Status()
	cstat.Volumes = stat.Volumes()
	cstat.Clients = stat.Clients()
	render_json(cstat, rw, req)
}
